package org.zjt.flink;

import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.operators.Order;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.io.jdbc.JDBCInputFormat;
import org.apache.flink.api.java.io.jdbc.JDBCOutputFormat;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.typeutils.RowTypeInfo;
import org.apache.flink.types.Row;
import org.apache.flink.util.Collector;

/**
 * Description:
 *
 * @author juntao.zhang
 * Date: 2018-10-09 下午4:53
 * @see
 */
@Slf4j
public class JdbcBatchJob {

    public static void main(String[] args) throws Exception {
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        // 从数据库中得到原数据
        DataSource<Row> input = env.createInput(
                JDBCInputFormat.buildJDBCInputFormat()
                        .setDrivername("com.mysql.jdbc.Driver")
                        .setDBUrl("jdbc:mysql://localhost:3306/test")
                        .setUsername("root")
                        .setPassword("123456")
                        .setQuery("select name, age from info")
                        .setRowTypeInfo(new RowTypeInfo(BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO))
                        .finish());


        // 输出统计结果到数据库中
        JDBCOutputFormat finish = JDBCOutputFormat.buildJDBCOutputFormat()
                .setDrivername("com.mysql.jdbc.Driver")
                .setDBUrl("jdbc:mysql://localhost:3306/test")
                .setUsername("root")
                .setPassword("123456")
                .setQuery("insert into info (name, age) values (?,?)")
                .finish();


        input.flatMap(new MyFlatMapFunction()).groupBy(0).sum(1)
                .sortPartition(1, Order.DESCENDING)
                .flatMap(new MyFlatMapFunction2()).output(finish);

        env.execute();
    }

    static class MyFlatMapFunction implements FlatMapFunction<Row,Tuple2<String,Integer>> {
        @Override
        public void flatMap(Row value, Collector<Tuple2<String, Integer>> out) throws Exception {
            for (int i = 0; i < value.getArity(); i++) {
                Object field = value.getField(i);
                log.info("field:",field);
            }
            out.collect(new Tuple2<>(String.valueOf(value.getField(0)),1));
        }
    }


    static class MyFlatMapFunction2 implements FlatMapFunction<Tuple2<String,Integer>,Row> {

        @Override
        public void flatMap(Tuple2<String, Integer> value, Collector<Row> out) throws Exception {
            Row row = new Row(2);
            row.setField(0,value.f0);
            row.setField(1,value.f1);
            out.collect(row);
        }
    }
}
